package cn.doitedu.api;


import beans.Person;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Arrays;
import java.util.List;

public class _03_Reduce_Demo {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        Person p1 = new Person("a", "male", 18);
        Person p2 = new Person("b", "male", 19);
        Person p3 = new Person("c", "male", 20);
        Person p4 = new Person("d", "female", 28);
        Person p5 = new Person("e", "female", 38);
        Person p6 = new Person("f", "female", 18);

        List<Person> list = Arrays.asList(p1, p2, p3, p4, p5, p6);

        DataStreamSource<Person> stream = env.fromCollection(list);

        stream.keyBy(Person::getGender)
                .reduce(new ReduceFunction<Person>() {
                    @Override
                    public Person reduce(Person value1, Person value2) throws Exception {
                        return value2.getAge()> value1.getAge()? value2:value1;
                    }
                })
                .print();


        env.execute();

    }


}



